package com.chance.cc.crawler.development.utils;

import com.alibaba.fastjson.JSON;
import com.chance.cc.crawler.core.record.CrawlerRequestRecord;
import org.apache.commons.lang3.StringUtils;
import redis.clients.jedis.*;

import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * created by CC on 2020/12/16
 * mail 279020185@qq.com
 */
public class RedisReader {

    String host;

    int port;

    int db;

    String user;

    String passwd;

    Jedis jedis;

    public RedisReader(String host,int port,int db){
        this.host = host;
        this.port = port;
        this.db = db;

        JedisPoolConfig config = new JedisPoolConfig();
        //控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；
        //如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
        config.setMaxTotal(1);
        //控制一个pool最多有多少个状态为idle(空闲的)的jedis实例。
        config.setMaxIdle(1);
        //表示当borrow(引入)一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；
        config.setMaxWaitMillis(3000);
        //在borrow一个jedis实例时，是否提前进行validate操作；如果为true，则得到的jedis实例均是可用的；
//        config.setTestOnBorrow(Boolean.valueOf(testOnBorrow));

        JedisPool jedisPool = new JedisPool(config, host, port, 30000);

        this.jedis=new Jedis(host,port,50000);
        this.jedis.select(db);
    }

    public String listPop(String queue){
        return this.jedis.lpop(queue);
    }

    public String mapPop(String queue,String key){
        String value = this.jedis.hget(queue, key);
        jedis.hdel(queue,key);
        return value;
    }

    public String mapGet(String queue,String key){
        String value = this.jedis.hget(queue, key);
        return value;
    }

    public CrawlerRequestRecord recordPop(String listQueue,String mapQueue){
        String key = listPop(listQueue);
        if (StringUtils.isBlank(key)){
            return null;
        }
        String val = mapPop(mapQueue,key);
        if (StringUtils.isNotBlank(val)){
            return JSON.parseObject(val,CrawlerRequestRecord.class);
        }
        return null;
    }

    public Long listPush(String listQueue, String value){
        return listPush(listQueue,new String[]{value});
    }
    public Long mapPush(String mapQueue, String key,String value){
        return jedis.hset(mapQueue,key,value);
    }

    public Long listPush(String listQueue, String... value){
        return jedis.lpush(listQueue,value);
    }
    public List<String> lrange(String listQueue, long start, long end){
        return jedis.lrange(listQueue,start,end);
    }

    public Long llen(String listQueue){
        return jedis.llen(listQueue);
    }
    public Long hlen(String mapQueue){
        return jedis.hlen(mapQueue);
    }

    public Long del(String... redisQueue){
        return jedis.del(redisQueue);
    }

    public void sadd(String redisQueue,String... members){
        jedis.sadd(redisQueue,members);
    }
    public boolean sismember(String redisQueue,String member){
        return jedis.sismember(redisQueue,member);
    }
    public Set<String> smembers(String redisQueue){
       return jedis.smembers(redisQueue);
    }

    private Jedis getJedis() {
        return jedis;
    }

    public static void main(String[] args) {
        RedisReader redisReader = new RedisReader("192.168.1.215", 6379, 4);
        Set<String> keys = redisReader.keys("*");
        Map<String, Float> keySize = new HashMap<>();
        long totalSize = 0;
        int processCount = 0;
        for (String key : keys) {
            try {
//                long statistics = statistics(key, redisReader);
//                keySize.put(key, ((float)statistics/1024/1024));

                String debug = redisReader.debug(key);
//
                String[] items = debug.split(" ");
                for (String item : items) {
                    if (item.trim().startsWith("serializedlength")){
                        Long size = Long.valueOf(item.split(":")[1].trim());
                        totalSize+=size;
                        System.out.println("debug key:"+key+",size: "+size);
                        keySize.put(key, ((float)size/1024/1024));
                        processCount++;
                        break;
                    }
                }
            } catch (Exception e) {
                System.out.println("error key:" + key);
            }
        }
        System.out.println("----------------------------------------");
        HashMap<String, Float> finalOut = new LinkedHashMap<>();
        keySize.entrySet()
                .stream()
                .sorted((p1, p2) -> p2.getValue().compareTo(p1.getValue()))
                .collect(Collectors.toList()).forEach(ele -> finalOut.put(ele.getKey(), ele.getValue()));
        finalOut.forEach((key,size) -> {
            System.out.println(key + "-->" + size + "M");
        });
        System.out.println("total key count:"+keys.size()+",process success count:" + processCount);
        System.out.println("total size :" + (float)totalSize/1024/1024/1024 + "G");
    }

    private static long statistics(String key,RedisReader redisReader) throws UnsupportedEncodingException {
        String type = redisReader.type(key);
        long length = 0L;
        switch (type){
            case "hash": length = hashStatistics(key,redisReader); break;
            case "list": length = listStatistics(key,redisReader); break;
            case "set": length = setStatistics(key,redisReader); break;
            //todo ...
        }
        return length;
    }

    private static long hashStatistics(String key,RedisReader redisReader) throws UnsupportedEncodingException {
        Jedis jedis = redisReader.getJedis();
        String scanCursor ="0";
        long length =0;
        do {
            ScanResult<Map.Entry<String, String>> scanResult = jedis.hscan(key, scanCursor);
            scanCursor = scanResult.getCursor();
            List<Map.Entry<String, String>> result = scanResult.getResult();
            for (Map.Entry<String, String> entry : result) {
                length += entry.getKey().getBytes("utf-8").length + entry.getValue().getBytes("utf-8").length;
            }
        } while (!scanCursor.equals("0"));

        return length;
    }

    private static long listStatistics(String key,RedisReader redisReader) throws UnsupportedEncodingException {
        Jedis jedis = redisReader.getJedis();
        Long llen = jedis.llen(key);
        long step = 100;
        long startCursor =0L;
        long endCursor = startCursor + step;
        long length =0;

        while (startCursor <= llen){
            List<String> values = jedis.lrange(key, startCursor, endCursor);
            for (String value : values) {
                length += value.getBytes("utf-8").length;
            }
            startCursor += step +1;
            endCursor = startCursor + step;
        }

        return length;
    }

    private static long setStatistics(String key,RedisReader redisReader) throws UnsupportedEncodingException {
        Jedis jedis = redisReader.getJedis();
        String scanCursor ="0";
        long length =0;
        do {
            ScanResult<String> scanResult = jedis.sscan(key, scanCursor);
            scanCursor = scanResult.getCursor();
            List<String> result = scanResult.getResult();
            for (String entry : result) {
                length += entry.getBytes("utf-8").length;
            }
        } while (!scanCursor.equals("0"));

        return length;
    }

    private String debug(String redisQueue){
        return jedis.debug(DebugParams.OBJECT(redisQueue));
    }

    public Set<String> keys(String pattern){
        return jedis.keys(pattern);
    }
    public Set<String> hKeys(String queueName){
        return jedis.hkeys(queueName);
    }

    private String type(String key){
        return jedis.type(key);
    }

}
